package com.zyx.flinkdemo.stream.sink;

import com.zyx.flinkdemo.pojo.Book;
import com.zyx.flinkdemo.stream.utils.ClickHouseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zyx
 * @since 2021/6/8 23:26
 * desc: 使用ClickHouse的JDBC写入ClickHouse案例
 */
@Slf4j
public class ClickHouseSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Book book1 = new Book();
        book1.setId(1004L);
        book1.setAuthors("zhangsan");

        Book book2 = new Book();
        book2.setId(1005L);
        book2.setYear(1991123123213123L);

        Book book3 = new Book();
        book3.setId(1006L);
        book3.setTitle("good book");

        DataStreamSource<Book> bookDataStreamSource = env
                .fromElements(
                         book1,
                         book2,
                         book3
                );

        bookDataStreamSource.addSink(ClickHouseUtil.getJdbcSink("insert into book values(?,?,?,?)"));

        bookDataStreamSource.print();


        // addSink(ClickHouseUtil.getJdbcSink("insert into book values(?,?,?,?)"));
        env.execute();
    }
}
